package cn.com.cennavi.service;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.Serializable;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;

import kafka.serializer.DefaultDecoder;
import kafka.serializer.StringDecoder;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaPairInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;

import azkaban.utils.Props;
import cn.com.cennavi.bean.IndexInputParmBean;
import cn.com.cennavi.config.KafkaConfigFile;
import cn.com.cennavi.config.SparkConfigFile;
import cn.com.cennavi.interfaces.IMineLabModel;
import cn.com.cennavi.sparkStreamService.IndexComputerAverageSpeedReduceBykeyAndWindowService;
import cn.com.cennavi.sparkStreamService.IndexReadCNTFDataRDDStream;
import cn.com.cennavi.sparkStreamService.IndexSendAverageSpeedTokafka;

public class IndexTaskSubmit extends IMineLabModel implements Serializable {

	/** @Fields serialVersionUID: */

	private static final long serialVersionUID = 1L;
	private static SparkConf sparkConf = null;
	private static Map<String, String> cityMeshIdMap = new HashMap<String, String>();
	static {
		cityMeshIdMap
				.put("110000",
						"595673,595663,595662,595673,595663,595672,605602,605614,605604,605613,605603,605605,595653,595665,595675,595674,595664,595673,595672,595654,595663,595660,595653,595652,595651,595650,595662,595661,595663,595670,595671,595662,595661,595672,95671,605611,605610,605600,595662,605602,605601,595661,595673,605612,605603,595672,595670,595671,595660,595577,605600,595567,605514,595564,605505,595573,595565,605601,605504,595574,605503,605516,595575,605515,595576,605506,605507,595554,595661,605517,595660,595566,595567,595564,595565,595557,595556,595555,595554,595536,595630,595631,595553,595535,595534,595533,595652,595651,595537,595650,595576,595621,595545,595544,595547,595546,595543,595525,595620,595640,595527,595526,595641,605604,595647,595635,605605,605606,595666,595665,595667,595676,595675,595644,595657,595674,595656,595664,595655,595636,595646,595654,595645,605624,605623,605636,605626,605635,605625,605634,605633,605615,605614,605604,605613,605603,605607,605605,605606,605616,605617,605627,605641,605527,605642,605537,605624,605611,605610,605623,605634,605620,605602,605633,605632,605516,605631,605622,605613,605612,605621,605630,605603,605517,595634,595621,595622,595635,595623,595632,595633,595624,595612,595631,595613,595653,595652,595642,595641,595644,595643,595645,595654,605642,605643,605644,605645,605663,605664,605665,615626,605625,615605,615645,615604,615606,615613,615632,615614,615633,615634,615635,605654,605655,605652,605653,605674,605675,605636,605635,615616,605634,615615,605633,615603,615624,615625,615622,615623,605732,605731,605730,605721,605720,605723,605741,605740,605722,605742,605637,605607,605617,605627,605710,605711,605712,605700,605701,605667,605741,605645,605646,605740,605665,605647,605760,605666,605761,605762,615703,615605,615607,615606,605655,605676,605771,605677,605772,605752,605656,605657,605675,605751,605770,605637,605750,605636,615617,605635,615616,605773,605774,615702,615701,615700,605627,605641,605642,605537,605556,605557,605663,605660,605661,605662,615604,605640,615612,615613,605652,605653,605546,605674,605547,605672,605673,605670,605671,605650,605651,605631,605630,615602,615603,615601,605567,605566,");
		cityMeshIdMap.put("310000", "466017,466027,466037,466047,466056,466057,466101,466102,466103,466110,466111,466112,466113,466114,466117,466120,466121,466122,466123,466124,466125,466126,466127,466130,466131,466132,466133,466134,466135,466136,466137,466140,466141,466142,466143,466144,466145,466146,466147,466150,466151,466152,466153,466154,466155,466156,466157,466160,466161,466162,466163,466164,466165,466166,466170,466171,466171,466172,466173,466174,466175,466176,466177,476100,476101,476102,476103,476104,476105,476106,476107,476111,476112,476113,476114,476115,476116,476117,476122,476123,476124,476125,476126,476127,476132,476133,476134,476135,476136,476141,476142,476143,476144,476145,476151,476152,476153,476161,476162,");
		cityMeshIdMap.put("510000", "345303,345312,345313,345322,345323,345324,345331,345332,345333,345334,345341,345342,345343,345344,345351,345352,345353,345354,345360,345361,345362,345363,345364,345370,345371,345372,345373,345374,355207,355217,355300,355301,355302,355303,355304,355310,355311,355312,355313,355320,355321,355322,355323,355331,355332,");
		cityMeshIdMap.put("518000", "335471,335460,335470,335461,335367,335366,335356,335377,335357,345306,345316,345307,345326,345317,335376,345327,345410,345400,345403,345412,345402,345411,345401,345413,335463,335454,335474,335464,335453,335473,335472,335462,");
	}
	
	/**
	 * azkaban固定写法
	 */
	private final String[] parameters;

	public IndexTaskSubmit(String name, Props props) throws Exception {
		this.parameters = props.getString("parameters").split(",");
	}

	public IndexTaskSubmit() {
		// TODO Auto-generated constructor stub
		this.parameters = null;
	}
	public void run() throws Exception {
		int[] inputDataIDs;
		int areaFlag;
		int[] areaCodes = new int[1];
		areaCodes[0] = 110000;
		//算法参数配置，这个在作为输入时应该写好，然后转化为字符串传进来
		Map<String, String> algoParam;
		String modelId = "indexModel";
		String modelminId = "indexAverageSpeed";
		String readtopic = "MQ1000002001";
		String outtopic = "testIndex001";
		String sparkTaskName = "indexRealTimeTaskSubmit_001";
//		String SparkStreamCheckPointPath1 = "file:///d:/data/111";
		String SparkStreamCheckPointPath1 = "hdfs://nujhadoop/test/minelab/indexCheckpoint/";
		
		// ------------------------------------------------------------------------//
//		System.setProperty("hadoop.home.dir", "d:\\hadoop-2.4.0");
		final String CITYMESHID = cityMeshIdMap.get(areaCodes[0] + "");

		// 读取默认的配置文件信息，得到spark连接
		SparkConfigFile sparkConfigFile = new SparkConfigFile();
		sparkConfigFile.init();
		sparkConf = sparkConfigFile.getSparkConf();
		Long time = sparkConfigFile.getSparkStreamDurationsSeconds();

		// 覆盖sparkstream默认参数
		sparkConf.setAppName(sparkTaskName);
		sparkConfigFile.setSparkStreamCheckPointPath(SparkStreamCheckPointPath1);

		// 建立sparkStream连接
		JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(time));
		System.out.println(sparkConfigFile.getSparkStreamCheckPointPath());
		ssc.checkpoint(sparkConfigFile.getSparkStreamCheckPointPath());

		// 覆盖kafka默认信息
		KafkaConfigFile kafkaConfigFile = new KafkaConfigFile();
		kafkaConfigFile.init();
		// 覆盖kafka默认信息
		HashSet<String> topicsSet1 = new HashSet<String>(Arrays.asList(readtopic.split(",")));
		kafkaConfigFile.setTopicsSet(topicsSet1);
		System.out.println(kafkaConfigFile.getKafkaParams() + "," + kafkaConfigFile.getTopicsSet());
		// 建立kafka连接
		JavaPairInputDStream<String, byte[]> messages = KafkaUtils.createDirectStream(ssc, String.class, byte[].class, StringDecoder.class, DefaultDecoder.class, kafkaConfigFile.getKafkaParams(), kafkaConfigFile.getTopicsSet());
		switch (modelminId) {
		case "indexAverageSpeed":
			System.out.println("indexAverageSpeed!");
			break;
		}
		// 读取cntf
		JavaPairDStream<Long, IndexInputParmBean> cntfLines = messages.flatMapToPair(new IndexReadCNTFDataRDDStream(CITYMESHID));
		// 滑动时间窗
		JavaPairDStream<Long, IndexInputParmBean> averageSpeedRdd = cntfLines.reduceByKeyAndWindow(new IndexComputerAverageSpeedReduceBykeyAndWindowService(), new Duration(sparkConfigFile.windowDuration), new Duration(sparkConfigFile.slideDuration), sparkConfigFile.runPartation);
		// 写kafka
		averageSpeedRdd.foreachRDD(new IndexSendAverageSpeedTokafka(kafkaConfigFile.metadata_broker_list, outtopic));
		ssc.start();
		ssc.awaitTermination();
//		ssc.close();
	}
	
	public static void main(String[] args) throws FileNotFoundException, IOException {
		// TODO Auto-generated method stub
		String userID;
		String password;
		int[] inputDataIDs;
		int areaFlag;
		int[] areaCodes = new int[1];
		areaCodes[0] = 110000;
		Map<String, String> algoParam;
		String modelId = "";
		String modelminId = "";
		String readtopic = "MQ1000002001";
		String outtopic = "testIndex001";
		String sparkTaskName = "indexRealTimeTaskSubmit_001";
//		String SparkStreamCheckPointPath1 = "file:///d:/data/111";
		String SparkStreamCheckPointPath1 = "hdfs://nujhadoop/test/minelab/indexCheckpoint/";
		
		// ------------------------------------------------------------------------//
//		System.setProperty("hadoop.home.dir", "d:\\hadoop-2.4.0");
		final String CITYMESHID = cityMeshIdMap.get(areaCodes[0] + "");

		// 读取默认的配置文件信息，得到spark连接
		SparkConfigFile sparkConfigFile = new SparkConfigFile();
		sparkConfigFile.init();
		sparkConf = sparkConfigFile.getSparkConf();
		Long time = sparkConfigFile.getSparkStreamDurationsSeconds();

		// 覆盖sparkstream默认参数
		sparkConf.setAppName(sparkTaskName);
		sparkConfigFile.setSparkStreamCheckPointPath(SparkStreamCheckPointPath1);

		// 建立sparkStream连接
		JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(time));
		System.out.println(sparkConfigFile.getSparkStreamCheckPointPath());
		ssc.checkpoint(sparkConfigFile.getSparkStreamCheckPointPath());

		// 覆盖kafka默认信息
		KafkaConfigFile kafkaConfigFile = new KafkaConfigFile();
		kafkaConfigFile.init();
		// 覆盖kafka默认信息
		HashSet<String> topicsSet1 = new HashSet<String>(Arrays.asList(readtopic.split(",")));
		kafkaConfigFile.setTopicsSet(topicsSet1);
		System.out.println(kafkaConfigFile.getKafkaParams() + "," + kafkaConfigFile.getTopicsSet());
		// 建立kafka连接
		JavaPairInputDStream<String, byte[]> messages = KafkaUtils.createDirectStream(ssc, String.class, byte[].class, StringDecoder.class, DefaultDecoder.class, kafkaConfigFile.getKafkaParams(), kafkaConfigFile.getTopicsSet());
		switch (modelminId) {
		case "indexAverageSpeed":
			System.out.println("indexAverageSpeed!");
			break;
		}
		// 读取cntf
		JavaPairDStream<Long, IndexInputParmBean> cntfLines = messages.flatMapToPair(new IndexReadCNTFDataRDDStream(CITYMESHID));
		// 滑动时间窗
		JavaPairDStream<Long, IndexInputParmBean> averageSpeedRdd = cntfLines.reduceByKeyAndWindow(new IndexComputerAverageSpeedReduceBykeyAndWindowService(), new Duration(sparkConfigFile.windowDuration), new Duration(sparkConfigFile.slideDuration), sparkConfigFile.runPartation);
		// 写kafka
		averageSpeedRdd.foreachRDD(new IndexSendAverageSpeedTokafka(kafkaConfigFile.metadata_broker_list, outtopic));
		ssc.start();
		ssc.awaitTermination();
//		ssc.close();
	}
}
